package cn.dfun.sample.flink.apitest
import org.apache.flink.api.common.functions.{FilterFunction, ReduceFunction, RichMapFunction}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._

/**
  * 模拟传感器求温度最小值
  */
object TransformTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val inputPath = "C:\\wor\\flink-sample\\src\\main\\resources\\sensor"
    val inputStream= env.readTextFile(inputPath)
    // 包装成样例类
    val dataStream = inputStream
      .map(data => {
        var arr = data.split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      })
    // 分组聚合
    val aggStream = dataStream
      .keyBy("id") // 根据id分组
      .minBy("temperature")
//    aggStream.print()

    // 当前最小温度值和最大时间戳
    val resultStream = dataStream
        .keyBy("id")
//        .reduce((curState, newData) => {
//          SensorReading(curState.id, newData.timestamp, curState.temperature.min(newData.temperature))
//        })
//      .filter(new MyFilter) // TODO
      .reduce(new MyReduceFunction)
//    resultStream.print()

    // 多流转换
    // 分流
    // 传感器数据分为高温和低温两个流
    val splitStream = dataStream
        .split(data => { // 弃用,推荐side output
          if(data.temperature > 30.0) Seq("high") else Seq("low")
        })
    val highStream = splitStream.select("high")
    val lowStream = splitStream.select("low")
    val allStream = splitStream.select("high", "low")
//    highStream.print("high")
//    lowStream.print("low")
//    allStream.print("all")

    // 合流
    val warningStream = highStream.map(data => (data.id, data.temperature))
    val connectedStream = warningStream.connect(lowStream)
    val comapResultStream = connectedStream
        .map(
          warningData => (warningData._1, warningData._2, "warning"),
          lowData => (lowData.id, "healthy")
        )
//    comapResultStream.print("comap")

    // union
    val unionStream = highStream.union(lowStream)

    env.execute("transform test")
  }
}

class MyReduceFunction extends ReduceFunction[SensorReading] {
  override def reduce(t: SensorReading, t1: SensorReading): SensorReading = {
    SensorReading(t.id, t.timestamp, t.temperature.min(t1.temperature))
  }
}

// 简单需求使用匿名函数即可,复杂需求包装成函数类更加直观
class MyFilter extends FilterFunction[SensorReading] {
  override def filter(t: SensorReading): Boolean = {
    t.id.startsWith("sensor_1") // 可以把过滤字符串作为构造函数参数
  }
}

// 与MapFunction的区别是,可以读写运行时上下文,还有一些生命周期方法
// 如根据数据库操作可以写在open方法,避免每次数据都建立连接
class MyRich extends RichMapFunction[SensorReading, String] {
  override def map(in: SensorReading): String = in.id + "temperature"

  override def open(parameters: Configuration): Unit = {
    getRuntimeContext
  }

  override def close(): Unit = {
    // 收尾工作,如关闭连接,清空状态
  }
}
